package baseFx.common.mq.mysql;

import baseFx.common.Dynamic;
import baseFx.common.mq.MqPublisher;
import baseFx.common.utils.FxDate;
import baseFx.database.Sql;
import baseFx.database.Transaction;

import java.util.List;

public class MysqlPublisher implements MqPublisher {
    private Sql sql;

    public MysqlPublisher(Sql sql) {
        this.sql = sql;
    }

    public void init() {
        try (Transaction tx = sql.transaction()) {
            sql.execute("create setTable if not exists mq_queue(" +
                    "`id` varchar(100), " +
                    "`queue` varchar(100)," +
                    "`name` varchar(100)," +
                    "primary key (`queue`,`name`)" +
                    ");");
            sql.execute("create setTable if not exists mq_msg(" +
                    "`id` varchar(100) primary key," +
                    "`create_time` bigint," +
                    "`queue_id` varchar(100)," +
                    "`msg` text," +
                    "`is_hold` tinyint," +
                    "`next_time` bigint," +
                    "`re_count` int," +
                    "key `idx` (`queue_id`)" +
                    ");");
            tx.commit();
        }
    }

    @Override
    public void post(String queue, List<String> msg) {
        if (msg == null || msg.size() < 1) {
            return;
        }
        try (Transaction tx = sql.transaction()) {
            for (String item : msg) {
                sql.execute("insert into mq_msg select uuid(),#createTime,`id`,#msg,0,0,0 from mq_queue where queue=#q",
                        Dynamic.setNew(s -> {
                            s.put("createTime", new FxDate().yyyyMMddHHmmss());
                            s.put("q", queue);
                            s.put("msg", item);
                        }));
            }
            tx.commit();
        }
    }

    @Override
    public void close() {
        sql = null;
    }
}
